package com.gsm.projects.jobDw;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.Properties;
import java.util.concurrent.Future;

public class MysqlToKafka {

    private static final String MYSQL_URL = "jdbc:mysql://localhost:3306/db_job_realtime_statistics";
    private static final String MYSQL_USER = "root";
    private static final String MYSQL_PASSWORD = "123456";
    private static final String KAFKA_BROKERS = "hadoop102:9092,hadoop103:9092,hadoop104:9092";
    private static final String KAFKA_TOPIC = "t_job_topic";

    public static void main(String[] args) {
        // 初始化Kafka Producer
        Properties props = new Properties();
        props.put("bootstrap.servers", KAFKA_BROKERS);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        Producer<String, String> producer = new KafkaProducer<>(props);

        // 连接MySQL数据库
        try (Connection conn = DriverManager.getConnection(MYSQL_URL, MYSQL_USER, MYSQL_PASSWORD)) {
            Statement stmt = conn.createStatement();
            ResultSet rs = stmt.executeQuery("SELECT * FROM t_job");

            // 遍历结果集并将数据发送到Kafka
            while (rs.next()) {
                int id = rs.getInt("id");
                String name = rs.getString("name");
                String city = rs.getString("city");
                String school = rs.getString("school");
                String type = rs.getString("type");
                String major = rs.getString("major");
                String graduationDate = rs.getDate("graduation_date").toString();
                String company = rs.getString("company");
                String position = rs.getString("position");
                double salary = rs.getDouble("salary");
                String receivedOfferTime = rs.getTimestamp("received_offer_time").toString();
                String entryDate = rs.getTimestamp("entry_date").toString();

                // 构建JSON格式的消息
                String message = String.format("{\"id\":%d,\"name\":\"%s\",\"city\":\"%s\",\"school\":\"%s\",\"type\":\"%s\",\"major\":\"%s\",\"graduation_date\":\"%s\",\"company\":\"%s\",\"position\":\"%s\",\"salary\":%.2f,\"received_offer_time\":\"%s\",\"entry_date\":\"%s\"}",
                        id, name, city, school, type, major, graduationDate, company, position, salary, receivedOfferTime, entryDate);

                // 发送消息到Kafka
                Future<RecordMetadata> future = producer.send(new ProducerRecord<>(KAFKA_TOPIC, String.valueOf(id), message));
                RecordMetadata metadata = future.get(); // 等待发送完成
                System.out.println("Sent message to topic " + metadata.topic() + " partition " + metadata.partition() + " offset " + metadata.offset());
            }

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (producer != null) {
                producer.close();
            }
        }
    }
}